package cn.doitedu.data_sync;

import cn.doitedu.udfs.GpsToGeohashFunction;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_GpsReference_To_Hbase {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);  // 以“批计算模式”来运行本job

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射mysql中的参考点表
        tenv.executeSql(
                "CREATE TABLE reference_gps_mysql (   " +
                        "    BD09_LAT   double, " +
                        "    BD09_LNG   double, " +
                        "    province string,   " +
                        "    city     string,   " +
                        "    region   string    " +
                        ") with (             " +
                        "    'connector' = 'jdbc',      " +
                        "    'url' = 'jdbc:mysql://doitedu:3306/realtimedw', " +
                        "    'table-name' = 'lv4_reference_gps', " +
                        "    'username' = 'root',       " +
                        "    'password' = 'root'        " +
                        ")");

        // 建表，映射hbase中的目标表
        tenv.executeSql(
                " CREATE TABLE ums_member_hbase (        "+
                        "  geohash STRING,               "+
                        "  f ROW<                        "+
                        "       province string,         "+
                        " 	    city string,             "+
                        " 	    region string            "+
                        " 	>,                           "+
                        "   primary key(geohash) not enforced	 "+
                        " ) WITH (                               "+
                        "  'connector' = 'hbase-2.2',            "+
                        "  'table-name' = 'dim_reference_geo',      "+
                        "  'zookeeper.quorum' = 'doitedu:2181'   "+
                        " )                                      "
        );



        // 注册gps转geohash的自定义函数
        tenv.createTemporaryFunction("gps2geo", GpsToGeohashFunction.class);

        tenv.executeSql(
                        " insert into ums_member_hbase                "+
                        " select                                      "+
                        "   gps2geo(BD09_LAT,BD09_LNG) as geohash,    "+
                        " 	row(province,city,region) as f            "+
                        " from reference_gps_mysql                    "
        );

    }

}
